﻿#region Copyright (c) 2013-04, Olaf Kober <amarok.projects@gmail.com>
//================================================================================
//	Permission is hereby granted, free of charge, to any person obtaining a copy
//	of this software and associated documentation files (the "Software"), to deal
//	in the Software without restriction, including without limitation the rights
//	to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//	copies of the Software, and to permit persons to whom the Software is
//	furnished to do so, subject to the following conditions:
//
//	The above copyright notice and this permission notice shall be included in
//	all copies or substantial portions of the Software.
//
//	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//	AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//	OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
//	THE SOFTWARE.
//================================================================================
#endregion

#if TESTS

using System;
using System.Threading;
using NUnit.Framework;


namespace Amarok.Agents
{
	[TestFixture]
	public class Test_DefaultMessageBus
	{
		private class TestMessage : Message
		{
		}

		private class TestInformation : Information
		{
		}

		private class TestOperation : Operation<String>
		{
		}

		private class AgentA : Agent
		{
			public AgentA(AgentEnvironment environment, AgentOptions options)
				: base(environment, options)
			{
			}

			public void TestPublish(Message message)
			{
				base.Publish(message);
			}
			public void TestPublish(Object label, Message message)
			{
				base.Publish(label, message);
			}
		}

		private class AgentB : Agent
		{
			public AgentB(AgentEnvironment environment, AgentOptions options)
				: base(environment, options)
			{
			}

			public Int32 MessageHandlerCalled;
			public Int32 InformationHandlerCalled;
			public Int32 OperationHandlerCalled;

			[MessageHandler, ConcurrentExecution, NoParallelism]
			private void _HandleMessage(TestMessage message)
			{
				MessageHandlerCalled++;
			}

			[InformationHandler, ConcurrentExecution, NoParallelism]
			private void _HandleInformation(TestInformation message)
			{
				InformationHandlerCalled++;
			}

			[OperationHandler, ConcurrentExecution, NoParallelism]
			private String _HandleOperation(TestOperation message)
			{
				OperationHandlerCalled++;
				message.ReportProgress(None.Instance);
				return "done";
			}
		}


		// --- TESTS ---


		[Test]
		public void Test_Message()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			messageBus.Connect(subscriber1);
			messageBus.Connect(subscriber2);

			var message = new TestMessage();
			publisher.TestPublish(message);

			Helper.WaitForCondition(5000, 10, () => subscriber1.MessageHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.MessageHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}

		[Test]
		public void Test_InformationMessage()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			messageBus.Connect(subscriber1);
			messageBus.Connect(subscriber2);

			var information = new TestInformation();
			publisher.TestPublish(information);

			Helper.WaitForCondition(5000, 10, () => subscriber1.InformationHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.InformationHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}

		[Test]
		public void Test_OperationMessage()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			messageBus.Connect(subscriber1);

			var operation = new TestOperation();
			publisher.TestPublish(operation);

			Helper.WaitForCondition(5000, 10, () => subscriber1.OperationHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(1));

			operation.Completion.Wait(5000);
			Assert.That(operation.Completion.Result, Is.EqualTo("done"));
		}

		[Test]
		public void Test_Connect_Without_Label_Message_With_Label()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			messageBus.Connect(subscriber1);
			messageBus.Connect(subscriber2);

			var message = new TestMessage();
			message.Label = "COM99";
			publisher.TestPublish(message);

			Helper.WaitForCondition(5000, 10, () => subscriber1.MessageHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.MessageHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}

		[Test]
		public void Test_Connect_With_Label_Message_Without_Label()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			messageBus.Connect("COM1", subscriber1);
			messageBus.Connect("COM2", subscriber2);

			var information = new TestInformation();
			information.ResetLabel();
			publisher.TestPublish(information);

			Helper.WaitForCondition(5000, 10, () => subscriber1.InformationHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.InformationHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}

		[Test]
		public void Test_Connect_With_Label_Message_With_Label()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			messageBus.Connect("COM1", subscriber1);
			messageBus.Connect("COM2", subscriber2);

			var information = new TestInformation();
			publisher.TestPublish("COM1", information);

			Helper.WaitForCondition(5000, 10, () => subscriber1.InformationHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(500, 10, () => subscriber2.InformationHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}

		[Test]
		public void Test_Connect_With_Label_Message_With_Label_Publisher_With_Label_Connected()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect("BLUB", publisher);
			messageBus.Connect("COM1", subscriber1);
			messageBus.Connect("COM2", subscriber2);

			var information = new TestInformation();
			publisher.TestPublish("COM1", information);

			Helper.WaitForCondition(5000, 10, () => subscriber1.InformationHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(500, 10, () => subscriber2.InformationHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}


		[Test]
		public void Test_Publish()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(subscriber1);
			messageBus.Connect(subscriber2);

			var message = new TestMessage();
			messageBus.Publish(message);

			Helper.WaitForCondition(5000, 10, () => subscriber1.MessageHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.MessageHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}

		[Test]
		public void Test_Publish_NullMessage()
		{
			var messageBus = new DefaultMessageBus();

			Assert.Throws<ArgumentNullException>(() =>
				{
					messageBus.Publish(null);
				});
		}

		[Test]
		public void Test_Publish_NullLabel()
		{
			var messageBus = new DefaultMessageBus();

			Assert.Throws<ArgumentNullException>(() =>
				{
					messageBus.Publish(null, new TestMessage());
				});
		}


		[Test]
		public void Test_PublishOperation()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var subscriber = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(subscriber);

			var operation = new TestOperation();
			var task = messageBus.PublishOperation(operation);
			task.Wait();

			Assert.That(task.Result, Is.EqualTo("done"));
			Assert.That(subscriber.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.OperationHandlerCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_PublishOperation_Label()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var subscriber = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(subscriber);

			var operation = new TestOperation();
			var task = messageBus.PublishOperation("COM1", operation);
			task.Wait();

			Assert.That(task.Result, Is.EqualTo("done"));
			Assert.That(subscriber.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.OperationHandlerCalled, Is.EqualTo(1));
			Assert.That(operation.Label, Is.EqualTo("COM1"));
		}

		[Test]
		public void Test_PublishOperation_Label_AlreadySet()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var subscriber = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(subscriber);

			var operation = new TestOperation();
			operation.Label = "COM99";
			var task = messageBus.PublishOperation("COM1", operation);
			task.Wait();

			Assert.That(task.Result, Is.EqualTo("done"));
			Assert.That(subscriber.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.OperationHandlerCalled, Is.EqualTo(1));
			Assert.That(operation.Label, Is.EqualTo("COM1"));
		}

		[Test]
		public void Test_PublishOperation_Progress()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var subscriber = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(subscriber);

			var progressCalled = 0;

			var operation = new TestOperation();
			var task = messageBus.PublishOperation(operation, progress => progressCalled++);
			task.Wait();

			Assert.That(task.Result, Is.EqualTo("done"));
			Assert.That(subscriber.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.OperationHandlerCalled, Is.EqualTo(1));
			Assert.That(progressCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_PublishOperation_Label_Progress()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var subscriber = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(subscriber);

			var progressCalled = 0;

			var operation = new TestOperation();
			var task = messageBus.PublishOperation("COM1", operation, progress => progressCalled++);
			task.Wait();

			Assert.That(task.Result, Is.EqualTo("done"));
			Assert.That(subscriber.MessageHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber.OperationHandlerCalled, Is.EqualTo(1));
			Assert.That(operation.Label, Is.EqualTo("COM1"));
			Assert.That(progressCalled, Is.EqualTo(1));
		}


		[Test]
		public void Test_Subscribe_Callback()
		{
			var messageBus = new DefaultMessageBus();

			var messageCallbackCalled = 0;
			var messageCallbackArgument = (Message)null;
			var messageCallbackSignal = new AutoResetEvent(false);

			var subscription = messageBus.Subscribe(typeof(Message),
				msg =>
				{
					messageCallbackCalled++;
					messageCallbackArgument = msg;
					messageCallbackSignal.Set();
				});

			var message = new TestOperation();
			messageBus.Publish(message);

			messageCallbackSignal.WaitOne(5000);
			Assert.That(messageCallbackCalled, Is.EqualTo(1));
			Assert.That(messageCallbackArgument, Is.SameAs(message));

			subscription.Dispose();
			messageBus.Publish(message);

			messageCallbackSignal.WaitOne(500);
			Assert.That(messageCallbackCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_Subscribe_Agent()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var messageBus = new DefaultMessageBus();
			var agent = new AgentB(environment, options);

			var subscription = messageBus.Subscribe(
				typeof(Message),
				agent.ToAgentReference());

			var message = new TestOperation();
			messageBus.Publish(message);

			Helper.WaitForCondition(5000, 10, () => agent.OperationHandlerCalled == 1);
			Assert.That(agent.OperationHandlerCalled, Is.EqualTo(1));

			subscription.Dispose();
			messageBus.Publish(message);

			Helper.WaitForCondition(500, 10, () => agent.OperationHandlerCalled == 2);
			Assert.That(agent.OperationHandlerCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_Subscribe_Callback_MessageFilter()
		{
			var messageBus = new DefaultMessageBus();

			var messageCallbackCalled = 0;
			var messageCallbackArgument = (Message)null;
			var messageCallbackSignal = new AutoResetEvent(false);

			var subscription = messageBus.Subscribe(typeof(Message),
				msg =>
				{
					messageCallbackCalled++;
					messageCallbackArgument = msg;
					messageCallbackSignal.Set();
				},
				msg =>
				{
					return "AAA".Equals(msg.Label);
				});

			var message = new TestOperation();
			messageBus.Publish(message);

			messageCallbackSignal.WaitOne(500);
			Assert.That(messageCallbackCalled, Is.EqualTo(0));

			message = new TestOperation();
			messageBus.Publish("AAA", message);

			messageCallbackSignal.WaitOne(5000);
			Assert.That(messageCallbackCalled, Is.EqualTo(1));
			Assert.That(messageCallbackArgument, Is.SameAs(message));

			subscription.Dispose();
			messageBus.Publish(message);

			messageCallbackSignal.WaitOne(500);
			Assert.That(messageCallbackCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_Subscribe_Agent_MessageFilter()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var messageBus = new DefaultMessageBus();
			var agent = new AgentB(environment, options);

			var subscription = messageBus.Subscribe(
				typeof(Message),
				agent.ToAgentReference(),
				msg =>
				{
					return "AAA".Equals(msg.Label);
				});

			var message = new TestOperation();
			messageBus.Publish(message);

			Helper.WaitForCondition(500, 10, () => agent.OperationHandlerCalled == 1);
			Assert.That(agent.OperationHandlerCalled, Is.EqualTo(0));

			message = new TestOperation();
			messageBus.Publish("AAA", message);

			Helper.WaitForCondition(5000, 10, () => agent.OperationHandlerCalled == 1);
			Assert.That(agent.OperationHandlerCalled, Is.EqualTo(1));

			subscription.Dispose();
			messageBus.Publish(message);

			Helper.WaitForCondition(500, 10, () => agent.OperationHandlerCalled == 2);
			Assert.That(agent.OperationHandlerCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_Subscribe_Callback_TypeSafe()
		{
			var messageBus = new DefaultMessageBus();

			var messageCallbackCalled = 0;
			var messageCallbackArgument = (TestInformation)null;
			var messageCallbackSignal = new AutoResetEvent(false);

			var subscription = messageBus.Subscribe<TestInformation>(
				msg =>
				{
					messageCallbackCalled++;
					messageCallbackArgument = msg;
					messageCallbackSignal.Set();
				});

			var message = new TestInformation();
			messageBus.Publish(message);

			messageCallbackSignal.WaitOne(5000);
			Assert.That(messageCallbackCalled, Is.EqualTo(1));
			Assert.That(messageCallbackArgument, Is.SameAs(message));

			subscription.Dispose();
			messageBus.Publish(message);

			messageCallbackSignal.WaitOne(500);
			Assert.That(messageCallbackCalled, Is.EqualTo(1));
		}

		[Test]
		public void Test_Subscribe_Callback_MessageFilter_TypeSafe()
		{
			var messageBus = new DefaultMessageBus();

			var messageCallbackCalled = 0;
			var messageCallbackArgument = (TestInformation)null;
			var messageCallbackSignal = new AutoResetEvent(false);

			var subscription = messageBus.Subscribe<TestInformation>(
				msg =>
				{
					messageCallbackCalled++;
					messageCallbackArgument = msg;
					messageCallbackSignal.Set();
				},
				msg =>
				{
					return "AAA".Equals(msg.Label);
				});

			var message = new TestInformation();
			messageBus.Publish(message);

			messageCallbackSignal.WaitOne(500);
			Assert.That(messageCallbackCalled, Is.EqualTo(0));

			message = new TestInformation();
			messageBus.Publish("AAA", message);

			messageCallbackSignal.WaitOne(5000);
			Assert.That(messageCallbackCalled, Is.EqualTo(1));
			Assert.That(messageCallbackArgument, Is.SameAs(message));

			subscription.Dispose();
			messageBus.Publish(message);

			messageCallbackSignal.WaitOne(500);
			Assert.That(messageCallbackCalled, Is.EqualTo(1));
		}


		[Test]
		public void Test_Disconnect_Dispose()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			var connection1 = messageBus.Connect(subscriber1);
			var connection2 = messageBus.Connect(subscriber2);

			var message = new TestMessage();
			publisher.TestPublish(message);

			Helper.WaitForCondition(5000, 10, () => subscriber1.MessageHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.MessageHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));

			connection1.Dispose();
			publisher.TestPublish(message);

			Helper.WaitForCondition(500, 10, () => subscriber1.MessageHandlerCalled == 2);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.MessageHandlerCalled == 2);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(2));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));

			connection2.Dispose();
			publisher.TestPublish(message);

			Helper.WaitForCondition(500, 10, () => subscriber1.MessageHandlerCalled == 2);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(500, 10, () => subscriber2.MessageHandlerCalled == 3);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(2));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}

		[Test]
		public void Test_Disconnect_Complete()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			var connection1 = messageBus.Connect(subscriber1);
			var connection2 = messageBus.Connect(subscriber2);

			var message = new TestMessage();
			publisher.TestPublish(message);

			Helper.WaitForCondition(5000, 10, () => subscriber1.MessageHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.MessageHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));

			TaskHelper.Run(() => connection1.Complete());
			TaskHelper.Run(() => connection1.Complete());
			connection1.Complete();
			publisher.TestPublish(message);
			connection1.Completion.Wait();

			Helper.WaitForCondition(500, 10, () => subscriber1.MessageHandlerCalled == 2);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.MessageHandlerCalled == 2);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(2));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));

			TaskHelper.Run(() => connection2.Complete());
			TaskHelper.Run(() => connection2.Complete());
			connection2.Complete();
			publisher.TestPublish(message);
			connection2.Completion.Wait();

			Helper.WaitForCondition(500, 10, () => subscriber1.MessageHandlerCalled == 2);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(500, 10, () => subscriber2.MessageHandlerCalled == 3);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(2));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}


		[Test]
		public void Test_Dispose()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			messageBus.Connect(subscriber1);
			messageBus.Connect(subscriber2);

			var message = new TestMessage();
			publisher.TestPublish(message);

			Helper.WaitForCondition(5000, 10, () => subscriber1.MessageHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.MessageHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));

			messageBus.Dispose();
			publisher.TestPublish(message);

			Helper.WaitForCondition(500, 10, () => subscriber1.MessageHandlerCalled == 2);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(500, 10, () => subscriber2.MessageHandlerCalled == 2);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}

		[Test]
		public void Test_Complete()
		{
			var environment = new AgentEnvironment();
			var options = AgentOptions.Empty;
			var publisher = new AgentA(environment, options);
			var subscriber1 = new AgentB(environment, options);
			var subscriber2 = new AgentB(environment, options);
			var messageBus = new DefaultMessageBus();

			messageBus.Connect(publisher);
			messageBus.Connect(subscriber1);
			messageBus.Connect(subscriber2);

			var message = new TestMessage();
			publisher.TestPublish(message);

			Helper.WaitForCondition(5000, 10, () => subscriber1.MessageHandlerCalled == 1);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(5000, 10, () => subscriber2.MessageHandlerCalled == 1);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));

			TaskHelper.Run(() => messageBus.Complete());
			TaskHelper.Run(() => messageBus.Complete());
			messageBus.Complete();
			publisher.TestPublish(message);
			messageBus.Completion.Wait();

			Helper.WaitForCondition(500, 10, () => subscriber1.MessageHandlerCalled == 2);
			Assert.That(subscriber1.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber1.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber1.OperationHandlerCalled, Is.EqualTo(0));

			Helper.WaitForCondition(500, 10, () => subscriber2.MessageHandlerCalled == 2);
			Assert.That(subscriber2.MessageHandlerCalled, Is.EqualTo(1));
			Assert.That(subscriber2.InformationHandlerCalled, Is.EqualTo(0));
			Assert.That(subscriber2.OperationHandlerCalled, Is.EqualTo(0));
		}

	}
}

#endif
